package cn.iocoder.yudao.module.forward.mqtt.broker;

//import cn.iocoder.yudao.module.forward.dal.dataobject.datasouce.DataSouceDO;

import cn.iocoder.yudao.framework.common.pojo.PageParam;
import cn.iocoder.yudao.framework.common.pojo.PageResult;
import cn.iocoder.yudao.module.forward.controller.admin.scope.vo.ScopePageReqVO;
import cn.iocoder.yudao.module.forward.dal.dataobject.scope.ScopeDO;
import cn.iocoder.yudao.module.forward.mqtt.*;
import cn.iocoder.yudao.module.forward.mqtt.MessageHandler;
import cn.iocoder.yudao.module.forward.service.scope.ScopeService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.HashMap;
import java.util.Map;

/**
 * 转发mqttInfo
 * topics 存放 key:topic   value:topicType
 * name为转发名称
 */
@Slf4j
@Service
public class BrokerMqttManager extends MqttManager {
    @Resource
    private ScopeService scopeService;
    @Resource(name = "BrokerMessageHandler")
    private MessageHandler handler;

    public BrokerMqttManager() {
        tag = "转发";
    }

    public void doConnect(MqttInfo mqttInfo) {
        if (mqttInfo == null) {
            return;
        }
        if (clientMaps.containsKey(mqttInfo.getName())) {
            log.error("内部错误, {}({})已存在", tag, mqttInfo.getName());
            return;
        }

        log.info("mqttInfo = {}", mqttInfo);
        // TODO: 2025/4/22
        //  回调函数类可单独处理，也可统一处理 单独处理时需要重新实现相关方法
//        MessageHandler handler = new BrokerMessageHandler(sourceMqttManager, scopeService, strategyService, eewService, eqrService);
        createClient(mqttInfo, new MqttCallBackImpl(this, handler, tag) {
            @Override
            public void subscribeTopics(Integer id) {
                ScopePageReqVO scopePageReqVO = new ScopePageReqVO();
                scopePageReqVO.setPageSize(PageParam.PAGE_SIZE_NONE);
                scopePageReqVO.setBrokerId(id);
                scopePageReqVO.setEnable(true);
                scopePageReqVO.setTopicType(MqttConstants.TOPIC_TYPE_ACK);
                PageResult<ScopeDO> scopePage = scopeService.getScopePage(scopePageReqVO);
                Map<String, String> topics = new HashMap<>();
                /**
                 * 由于授权可能存在多个一个broker下不同分组下的topic相同，所以最终的map存放的是topic的并集
                 */
                for (ScopeDO scopeDO : scopePage.getList()) {
                    if (!scopeDO.getTopicType().equals(MqttConstants.TOPIC_TYPE_ACK)) {
                        continue;
                    }
                    topics.put(scopeDO.getTopic(), scopeDO.getTopicType());
                }
                if (topics != null && topics.size() > 0) {
                    MqttConnect mqttConnect = getMqttConnect(mqttInfo.getName());
                    for (String topic : topics.keySet()) {
                        mqttConnect.subscribe(topic);
                    }
                    mqttInfo.getTopics().putAll(topics);
                }
            }
        });
    }

//    @Override
//    public MqttInfo subscribe(String name, String topic, Object obj) {
//        MqttInfo mqttInfo = super.subscribe(name, topic, obj);
//        if (mqttInfo != null) {
//            mqttInfo.getTopics().put(topic, obj);
//            return mqttInfo;
//        }
//        return null;
//    }
//
//    @Override
//    public MqttInfo unSubscribe(String name, String topic, Object obj) {
//        MqttInfo mqttInfo = super.unSubscribe(name, topic, obj);
//        if (mqttInfo != null) {
//            mqttInfo.getTopics().remove(topic);
//            return mqttInfo;
//        }
//        return null;
//    }
}
